package com.sangsang.kafka.config;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.kafka.config.AbstractKafkaListenerContainerFactory;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.config.KafkaListenerContainerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.stereotype.Component;

import java.util.Map;

/**
 * @author liutangqi
 * @date 2023/12/21 16:25
 */
@Component
@ConfigurationProperties(prefix = "spring.kafka")
public class KafkaConfiguration {
    public static final String SINGLE_CONSUME_BATCH_FACTORY = "singleConsumeBatchFactory";
    private static final Logger log = LoggerFactory.getLogger(KafkaConfiguration.class);

    /**
     * 单个消费者，批量消费
     *
     * @author liutangqi
     * @date 2023/12/21 16:27
     * @Param [kafkaListenerContainerFactory]
     **/
    @Bean(name = SINGLE_CONSUME_BATCH_FACTORY)
    public KafkaListenerContainerFactory<?> traceBatchConsumeFactory(AbstractKafkaListenerContainerFactory kafkaListenerContainerFactory) {
        log.info("开始初始化kafka消费者工厂");
        // 批量消费配置
        ConcurrentKafkaListenerContainerFactory<Integer, String> batchConsumeFactory = new ConcurrentKafkaListenerContainerFactory<>();
        // 消费者配置
        Map<String, Object> consumerProperties = kafkaListenerContainerFactory.getConsumerFactory().getConfigurationProperties();
        // 消费工厂
        batchConsumeFactory.setConsumerFactory(new DefaultKafkaConsumerFactory<>(consumerProperties));
        // 设置为批量消费，每个批次数量在Kafka配置参数中设置ConsumerConfig.MAX_POLL_RECORDS_CONFIG
        batchConsumeFactory.setBatchListener(true);
        //单个消费者
        batchConsumeFactory.setConcurrency(1);
        return batchConsumeFactory;
    }
}
